package cn.com.sparkle.paxos.net.callback;

import java.util.concurrent.locks.ReentrantLock;

import org.apache.log4j.Logger;

import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.MessagePackage;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.VoteResponse;
import cn.com.sparkle.paxos.net.CallBack;
import cn.com.sparkle.paxos.net.NetNode;
//import cn.com.sparkle.paxos.net.message.VoteResponse;
import cn.com.sparkle.paxos.paxosinstance.PaxosInstance;

public class VoteCallBack implements CallBack {
	private final static Logger logger = Logger.getLogger(VoteCallBack.class);

	private int needResponseCount;
	private final int quorum;
	private int goodNum = 0;
	private int badNum = 0;

	private PaxosInstance paxosInstance;
	public volatile long refuseId = -1;

	private ReentrantLock lock = new ReentrantLock();

	public VoteCallBack(int quorum, int needResponseCount,
			PaxosInstance paxosInstance) {
		this.needResponseCount = needResponseCount;
		this.quorum = quorum;
		this.paxosInstance = paxosInstance;
	}

	@Override
	public void call(NetNode nnode, MessagePackage value) {
		try {
			lock.lock();
			VoteResponse response = value.getVoteResponse();
			if (response.getRefuseId() == -1) {
				++goodNum;
			} else {
				if (response.getRefuseId() > refuseId) {
					refuseId = response.getRefuseId();
				}
				++badNum;
			}
			--needResponseCount;
			checkCount();
		} finally {
			lock.unlock();
		}
	}

	private void checkCount() {
		logger.debug("goodNum:" + goodNum + " badNum:" + badNum
				+ " needResponseCount:" + needResponseCount);
		if (goodNum == quorum) {
			++goodNum;// avoid invoking repeatedly
			paxosInstance.voteSuccess();
		} else if (goodNum < quorum && needResponseCount == 0) {
			if (badNum >= quorum) {
				paxosInstance.fail(refuseId);
			} else {
				paxosInstance.fail(-1);
			}
		}
	}

	@Override
	public void fail(NetNode nnode) {
		try {
			lock.lock();
			--needResponseCount;
			checkCount();
		} finally {
			lock.unlock();
		}
	}

}
